/*------------------------------------------------------------------------------
* stream.c : stream input/output functions
*
*          Copyright (C) 2008-2016 by T.TAKASU, All rights reserved.
*
* options : -DWIN32    use WIN32 API
*           -DSVR_REUSEADDR reuse tcp server address
*
* references :
*     [1] RTCM Recommendaed Standards for Networked Transport for RTCM via
*         Internet Protocol (Ntrip), Version 1.0, Semptember 30, 2004
*     [2] H.Niksic and others, GNU Wget 1.12, The non-iteractive download
*         utility, 4 September 2009
*
* version : $Revision:$ $Date:$
* history : 2009/01/16 1.0  new
*           2009/04/02 1.1  support nmea request in ntrip request
*                           support time-tag of file as stream
*           2009/09/04 1.2  ported to linux environment
*                           add fflush() to save file stream
*           2009/10/10 1.3  support multiple connection for tcp server
*                           add keyword replacement in file path
*                           add function strsendnmea(), strsendcmd()
*           2010/07/18 1.4  support ftp/http stream types
*                           add keywords replacement of %ha,%hb,%hc in path
*                           add api: strsetdir(),strsettimeout()
*           2010/08/31 1.5  reconnect after error of ntrip client
*                           fix bug on no file swap at week start (2.4.0_p6)
*           2011/05/29 1.6  add fast stream replay mode
*                           add time margin to swap file
*                           change api strsetopt()
*                           introduce non_block send for send socket
*                           add api: strsetproxy()
*           2011/12/21 1.7  fix bug decode tcppath (rtklib_2.4.1_p5)
*           2012/06/09 1.8  fix problem if user or password contains /
*                           (rtklib_2.4.1_p7)
*           2012/12/25 1.9  compile option SVR_REUSEADDR added
*           2013/03/10 1.10 fix problem with ntrip mountpoint containing "/"
*           2013/04/15 1.11 fix bug on swapping files if swapmargin=0
*           2013/05/28 1.12 fix bug on playback of file with 64 bit size_t
*           2014/05/23 1.13 retry to connect after gethostbyname() error
*                           fix bug on malloc size in openftp()
*           2014/06/21 1.14 add general hex message rcv command by !HEX ...
*           2014/10/16 1.15 support stdin/stdou for input/output from/to file
*           2014/11/08 1.16 fix getconfig error (87) with bluetooth device
*           2015/01/12 1.15 add rcv command to change bitrate by !BRATE
*           2016/01/16 1.16 add constant CRTSCTS for non-CRTSCTS-defined env.
*                           fix serial status for non-windows systems
*           2016/06/09 1.17 fix bug on !BRATE rcv command always failed
*                           fix program on struct alignment in time tag header
*           2016/06/21 1.18 reverse time-tag handler of file to previous
*           2016/07/23 1.19 add output of received stream to tcp port for serial
*           2016/08/20 1.20 modify api strsendnmea()
*           2016/08/29 1.21 fix bug on starting serial thread for windows
*           2016/09/03 1.22 add ntrip caster functions
*                           add api strstatx(),strsetsrctbl()
*                           add api strsetsel(),strgetsel()
*           2016/09/06 1.23 fix bug on ntrip caster socket and request handling
*           2016/09/27 1.24 support udp server and client
*           2016/10/10 1.25 support ::P={4|8} option in path for STR_FILE
*-----------------------------------------------------------------------------*/
#include <ctype.h>
#include "rtklib_fun.h"
#ifndef WIN32
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/time.h>
#define __USE_MISC
#ifndef CRTSCTS
#define CRTSCTS  020000000000
#endif
#include <errno.h>
#include <termios.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#endif

/* constants -----------------------------------------------------------------*/

#define def_thread_t pthread_t
#define def_dev_t int
#define def_cond_t pthread_cond_t
#define def_lock_t pthread_mutex_t
#define def_initlock(f) pthread_mutex_init(f, NULL)
#define def_lock(f) pthread_mutex_lock(f)
#define def_unlock(f) pthread_mutex_unlock(f)
#define def_destroylock(f) pthread_mutex_destroy(f)

#define TINTACT             200         /* period for stream active (ms) */
#define SERIBUFFSIZE        4096        /* serial buffer size (bytes) */
#define TIMETAGH_LEN        64          /* time tag file header length */
#define MAXCLI              32          /* max client connection for tcp svr */
#define MAXSTATMSG          32          /* max length of status message */
#define DEFAULT_MEMBUF_SIZE 4096        /* default memory buffer size (bytes) */

#define NTRIP_AGENT         "RTKLIB/" VER_RTKLIB
#define NTRIP_CLI_PORT      2101        /* default ntrip-client connection port */
#define NTRIP_SVR_PORT      80          /* default ntrip-server connection port */
#define NTRIP_MAXRSP        32768       /* max size of ntrip response */
#define NTRIP_MAXSTR        256         /* max length of mountpoint string */
#define NTRIP_RSP_OK_CLI    "ICY 200 OK\r\n" /* ntrip response: client */
#define NTRIP_RSP_OK_SVR    "OK\r\n"    /* ntrip response: server */
#define NTRIP_RSP_SRCTBL    "SOURCETABLE 200 OK\r\n" /* ntrip response: source table */
#define NTRIP_RSP_TBLEND    "ENDSOURCETABLE"
#define NTRIP_RSP_HTTP      "HTTP/"     /* ntrip response: http */
#define NTRIP_RSP_ERROR     "ERROR"     /* ntrip response: error */
#define NTRIP_RSP_UNAUTH    "HTTP/1.0 401 Unauthorized\r\n"
#define NTRIP_RSP_ERR_PWD   "ERROR - Bad Pasword\r\n"
#define NTRIP_RSP_ERR_MNTP  "ERROR - Bad Mountpoint\r\n"

#define FTP_CMD             "wget"      /* ftp/http command */
#define FTP_TIMEOUT         30          /* ftp/http timeout (s) */

#define MIN(x,y)            ((x)<(y)?(x):(y))

/* macros --------------------------------------------------------------------*/

#ifdef WIN32
#define dev_t               HANDLE
#define socket_t            SOCKET
#define closesocket         close
typedef int socklen_t;
#else
#define dev_t               int
#define socket_t            int
#define closesocket         close
#endif


typedef struct {            /* tcp control type */
    int state;              /* state (0:close,1:wait,2:connect) */
    char saddr[256];        /* address string */
    int port;               /* port */
    struct sockaddr_in addr; /* address resolved */
    socket_t sock;          /* socket descriptor */
    int tcon;               /* reconnect time (ms) (-1:never,0:now) */
    unsigned int tact;      /* data active tick */
    unsigned int tdis;      /* disconnect tick */

} tcp_t;
typedef struct {            /* tcp cilent type */
    tcp_t svr;              /* tcp server control */
    int toinact;            /* inactive timeout (ms) (0:no timeout) */
    int tirecon;            /* reconnect interval (ms) (0:no reconnect) */
} tcpcli_t;
typedef struct {            /* ntrip control type */
    int state;              /* state (0:close,1:wait,2:connect) */
    int type;               /* type (0:server,1:client) */
    int nb;                 /* response buffer size */
    char url[256];          /* url for proxy */
    char mntpnt[256];       /* mountpoint */
    char user[256];         /* user */
    char passwd[256];       /* password */
    char str[NTRIP_MAXSTR]; /* mountpoint string for server */
    unsigned char buff[NTRIP_MAXRSP]; /* response buffer */
    tcpcli_t* tcp;          /* tcp client */
} ntrip_t;

/* global options ------------------------------------------------------------*/

static int toinact = 1000000; /* inactive timeout (ms) */
static int ticonnect = 10000; /* interval to re-connect (ms) */
static int tirate = 1000;  /* avraging time for data rate (ms) */
static int buffsize = 32768; /* receive/send buffer size (bytes) */
static char localdir[1024] = ""; /* local directory for ftp/http */
static char proxyaddr[256] = ""; /* http/ntrip/ftp proxy address */
static unsigned int tick_master = 0; /* time tick master for replay */
static int fswapmargin = 30;  /* file swap margin (s) */
static int currentMsgCount = 0; /* self defined */

/* decode tcp/ntrip path (path=[user[:passwd]@]addr[:port][/mntpnt[:str]]) ---*/
void decodetcppath(const char* path, char* addr, char* port, char* user,
                   char* passwd, char* mntpnt, char* str) 	{
    char buff[MAXSTRPATH], * p, * q;

    if (port) *port = '\0';
    if (user) *user = '\0';
    if (passwd) *passwd = '\0';
    if (mntpnt) *mntpnt = '\0';
    if (str) *str = '\0';

    strcpy(buff, path);

    if (!(p = strrchr(buff, '@'))) p = buff;

    if ((p = strchr(p, '/')))
    {
        if ((q = strchr(p + 1, ':')))
        {
            *q = '\0'; if (str) strcpy(str, q + 1);
        }
        *p = '\0'; if (mntpnt) strcpy(mntpnt, p + 1);
    }
    if ((p = strrchr(buff, '@')))
    {
        *p++ = '\0';
        if ((q = strchr(buff, ':')))
        {
            *q = '\0'; if (passwd) strcpy(passwd, q + 1);
        }
        if (user) strcpy(user, buff);
    }
    else p = buff;

    if ((q = strchr(p, ':')))
    {
        *q = '\0'; if (port) strcpy(port, q + 1);
    }
    if (addr) strcpy(addr, p);
}
/*#ifdef WIN32
static int errsock(void) { return WSAGetLastError(); }
#else*/
/*static int errsock(void) { return errno; }*/
//#endif
/*static void discontcp(tcp_t* tcp, int tcon) 	{
    tracet(3, "discontcp: sock=%d tcon=%d\n", tcp->sock, tcon);

    closesocket(tcp->sock);
    tcp->state = 0;
    tcp->tcon = tcon;
    tcp->tdis = tickget();
}*/
/*static int recv_nb(socket_t sock, unsigned char* buff, int n) 	{
    struct timeval tv = { 0 };
    fd_set rs;
    int ret, nr;

    FD_ZERO(&rs); FD_SET(sock, &rs);
    ret = select(sock + 1, &rs, NULL, NULL, &tv);
    if (ret <= 0) return ret;
    nr = recv(sock, (char*)buff, n, 0);
    return nr <= 0 ? -1 : nr;
}*/
/*static int setsock(socket_t sock, char* msg) 	{
    int bs = buffsize, mode = 1;
#ifdef WIN32
    int tv = 0;
#else
    struct timeval tv = { 0 };
#endif
    tracet(3, "setsock: sock=%d\n", sock);

    if (setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof(tv)) == -1 ||
        setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (const char*)&tv, sizeof(tv)) == -1)
    {
        sprintf(msg, "sockopt error: notimeo");
        tracet(1, "setsock: setsockopt error 1 sock=%d err=%d\n", sock, errsock());
        closesocket(sock);
        return 0;
    }
    if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (const char*)&bs, sizeof(bs)) == -1 ||
        setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (const char*)&bs, sizeof(bs)) == -1)
    {
        tracet(1, "setsock: setsockopt error 2 sock=%d err=%d bs=%d\n", sock, errsock(), bs);
        sprintf(msg, "sockopt error: bufsiz");
    }
    if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (const char*)&mode, sizeof(mode)) == -1)
    {
        tracet(1, "setsock: setsockopt error 3 sock=%d err=%d\n", sock, errsock());
        sprintf(msg, "sockopt error: nodelay");
    }
    return 1;
}*/
/*static int gentcp(tcp_t* tcp, int type, char* msg) 	{
    struct hostent* hp;
#ifdef SVR_REUSEADDR
    int opt = 1;
#endif

    tracet(3, "gentcp: type=%d\n", type);

    *//* generate socket *//*
    if ((tcp->sock = socket(AF_INET, SOCK_STREAM, 0)) == (socket_t)-1)
    {
        sprintf(msg, "socket error (%d)", errsock());
        tracet(1, "gentcp: socket error err=%d\n", errsock());
        tcp->state = -1;
        return 0;
    }
    if (!setsock(tcp->sock, msg))
    {
        tcp->state = -1;
        return 0;
    }
    memset(&tcp->addr, 0, sizeof(tcp->addr));
    tcp->addr.sin_family = AF_INET;
    tcp->addr.sin_port = htons(tcp->port);

    if (type == 0)
    { *//* server socket *//*

#ifdef SVR_REUSEADDR
        *//* multiple-use of server socket *//*
			setsockopt(tcp->sock, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt,
				sizeof(opt));
#endif
        if (bind(tcp->sock, (struct sockaddr*)&tcp->addr, sizeof(tcp->addr)) == -1)
        {
            sprintf(msg, "bind error (%d) : %d", errsock(), tcp->port);
            tracet(1, "gentcp: bind error port=%d err=%d\n", tcp->port, errsock());
            closesocket(tcp->sock);
            tcp->state = -1;
            return 0;
        }
        listen(tcp->sock, 5);
    }
    else
    { *//* client socket *//*
        if (!(hp = gethostbyname(tcp->saddr)))
        {
            sprintf(msg, "address error (%s)", tcp->saddr);
            tracet(1, "gentcp: gethostbyname error addr=%s err=%d\n", tcp->saddr, errsock());
            closesocket(tcp->sock);
            tcp->state = 0;
            tcp->tcon = ticonnect;
            tcp->tdis = tickget();
            return 0;
        }
        memcpy(&tcp->addr.sin_addr, hp->h_addr, hp->h_length);
    }
    tcp->state = 1;
    tcp->tact = tickget();
    tracet(5, "gentcp: exit sock=%d\n", tcp->sock);
    return 1;
}*/
/*static int connect_nb(socket_t sock, struct sockaddr* addr, socklen_t len) 	{
#ifdef WIN32
    u_long mode = 1;
		int err;

		ioctlsocket(sock, FIONBIO, &mode);
		if (connect(sock, addr, len) == -1)
		{
			err = errsock();
			if (err == WSAEWOULDBLOCK || err == WSAEINPROGRESS ||
				err == WSAEALREADY || err == WSAEINVAL) return 0;
			if (err != WSAEISCONN) return -1;
		}
#else
    struct timeval tv = { 0 };
    fd_set rs, ws;
    int err, flag;

    flag = fcntl(sock, F_GETFL, 0);
    fcntl(sock, F_SETFL, flag | O_NONBLOCK);
    if (connect(sock, addr, len) == -1)
    {
        err = errsock();
        if (err != EISCONN && err != EINPROGRESS && err != EALREADY) return -1;
        FD_ZERO(&rs); FD_SET(sock, &rs); ws = rs;
        if (select(sock + 1, &rs, &ws, NULL, &tv) == 0) return 0;
    }
#endif
    return 1;
}*/
/*static int consock(tcpcli_t* tcpcli, char* msg) 	{
    int stat, err;

    tracet(4, "consock: sock=%d\n", tcpcli->svr.sock);

    *//* wait re-connect *//*
    if (tcpcli->svr.tcon < 0 || (tcpcli->svr.tcon > 0 &&
                                 (int)(tickget() - tcpcli->svr.tdis) < tcpcli->svr.tcon))
    {
        return 0;
    }
    *//* non-block connect *//*
    if ((stat = connect_nb(tcpcli->svr.sock, (struct sockaddr*)&tcpcli->svr.addr,
                           sizeof(tcpcli->svr.addr))) == -1)
    {
        err = errsock();
        sprintf(msg, "connect error (%d)", err);
        tracet(1, "consock: connect error sock=%d err=%d\n", tcpcli->svr.sock, err);
        closesocket(tcpcli->svr.sock);
        tcpcli->svr.state = 0;
        return 0;
    }
    if (!stat)
    { *//* not connect *//*
        sprintf(msg, "connecting...");
        return 0;
    }
    sprintf(msg, "%s", tcpcli->svr.saddr);
    tracet(2, "consock: connected sock=%d addr=%s\n", tcpcli->svr.sock, tcpcli->svr.saddr);
    tcpcli->svr.state = 2;
    tcpcli->svr.tact = tickget();
    return 1;
}*/
/*static int waittcpcli(tcpcli_t* tcpcli, char* msg) 	{
    tracet(4, "waittcpcli: sock=%d state=%d\n", tcpcli->svr.sock, tcpcli->svr.state);

    if (tcpcli->svr.state < 0) return 0;

    if (tcpcli->svr.state == 0)
    { *//* close *//*
        if (!gentcp(&tcpcli->svr, 1, msg)) return 0;
    }
    if (tcpcli->svr.state == 1)
    { *//* wait *//*
        if (!consock(tcpcli, msg)) return 0;
    }
    if (tcpcli->svr.state == 2)
    { *//* connect *//*
        if (tcpcli->toinact > 0 &&
            (int)(tickget() - tcpcli->svr.tact) > tcpcli->toinact)
        {
            sprintf(msg, "timeout");
            tracet(2, "waittcpcli: inactive timeout sock=%d\n", tcpcli->svr.sock);
            discontcp(&tcpcli->svr, tcpcli->tirecon);
            return 0;
        }
    }
    return 1;
}*/
/*static int send_nb(socket_t sock, unsigned char* buff, int n) 	{
    struct timeval tv = { 0 };
    fd_set ws;
    int ret, ns;

    FD_ZERO(&ws); FD_SET(sock, &ws);
    ret = select(sock + 1, NULL, &ws, NULL, &tv);
    if (ret <= 0) return ret;
    ns = send(sock, (char*)buff, n, 0);
    return ns < n ? -1 : ns;
}*/

/* open tcp client -----------------------------------------------------------*/
static tcpcli_t* opentcpcli(const char* path, char* msg) 	{
    tcpcli_t* tcpcli, tcpcli0 = { {0} };
    char port[256] = "";

    if (!(tcpcli = (tcpcli_t*)malloc(sizeof(tcpcli_t)))) return NULL;
    *tcpcli = tcpcli0;
    decodetcppath(path, tcpcli->svr.saddr, port, NULL, NULL, NULL, NULL);
    if (sscanf(port, "%d", &tcpcli->svr.port) < 1)
    {
        sprintf(msg, "port error: %s", port);
        free(tcpcli);
        return NULL;
    }
    tcpcli->svr.tcon = 0;
    tcpcli->toinact = toinact;
    tcpcli->tirecon = ticonnect;
    return tcpcli;
}
/* close tcp client ----------------------------------------------------------*/
static void closetcpcli(tcpcli_t* tcpcli) 	{
    closesocket(tcpcli->svr.sock);
    free(tcpcli);
}
static ntrip_t* openntrip(const char* path, int type, char* msg) 	{
    ntrip_t* ntrip;
    int i;
    char addr[256] = "", port[256] = "", tpath[MAXSTRPATH];

    if (!(ntrip = (ntrip_t*)malloc(sizeof(ntrip_t)))) return NULL;

    ntrip->state = 0;
    ntrip->type = type; /* 0:server,1:client */
    ntrip->nb = 0;
    ntrip->url[0] = '\0';
    ntrip->mntpnt[0] = ntrip->user[0] = ntrip->passwd[0] = ntrip->str[0] = '\0';
    for (i = 0; i < NTRIP_MAXRSP; i++)
        ntrip->buff[i] = 0;

    /* decode tcp/ntrip path */
    decodetcppath(path, addr, port, ntrip->user, ntrip->passwd, ntrip->mntpnt,
                  ntrip->str);

    /* use default port if no port specified */
    if (!*port)
    {
        sprintf(port, "%d", type ? NTRIP_CLI_PORT : NTRIP_SVR_PORT);
    }
    sprintf(tpath, "%s:%s", addr, port);

    /* ntrip access via proxy server */
    if (*proxyaddr)
    {
        sprintf(ntrip->url, "http://%s", tpath);
        strcpy(tpath, proxyaddr);
    }
    /* open tcp client stream */
    if (!(ntrip->tcp = opentcpcli(tpath, msg)))
    {
        free(ntrip);
        return NULL;
    }
    return ntrip;
}
/* close ntrip ---------------------------------------------------------------*/
static void closentrip(ntrip_t* ntrip) 	{
    closetcpcli(ntrip->tcp);
    free(ntrip);
}
extern int stropen(stream_t * stream, int type, int mode, const char* path) 	{
    stream->type = type;
    stream->mode = mode;
    strcpy(stream->path, path);
    stream->inb = stream->inr = stream->outb = stream->outr = 0;
    stream->tick_i = stream->tick_o = tickget();
    stream->inbt = stream->outbt = 0;
    stream->msg[0] = '\0';
    stream->port = NULL;
    switch (type)
    {
        //case STR_SERIAL: stream->port = openserial(path, mode, stream->msg); break;
        //case STR_FILE: stream->port = openfile(path, mode, stream->msg); break;
        //case STR_TCPSVR: stream->port = opentcpsvr(path, stream->msg); break;
        case STR_TCPCLI: stream->port = opentcpcli(path, stream->msg); break;
        case STR_NTRIPSVR: stream->port = openntrip(path, 0, stream->msg); break;
        case STR_NTRIPCLI: stream->port = openntrip(path, 1, stream->msg); break;
            //case STR_NTRIPC_S: stream->port = openntripc(path, 0, stream->msg); break;
            //case STR_NTRIPC_C: stream->port = openntripc(path, 1, stream->msg); break;
            //case STR_UDPSVR: stream->port = openudpsvr(path, stream->msg); break;
            //case STR_UDPCLI: stream->port = openudpcli(path, stream->msg); break;
            //case STR_MEMBUF: stream->port = openmembuf(path, stream->msg); break;
            //case STR_FTP: stream->port = openftp(path, 0, stream->msg); break;
            //case STR_HTTP: stream->port = openftp(path, 1, stream->msg); break;
            //case STR_VRSCONCLI: stream->port = openvrsconcli(path, stream->msg); break;
            //case STR_VRSCONSVR: stream->port = openvrsconsvr(path, stream->msg); break;
        default: stream->state = 0; return 1;
    }
    stream->state = !stream->port ? -1 : 1;
    return stream->port != NULL;
}
extern void strinit(stream_t * stream) 	{
    tracet(3, "strinit:\n");

    stream->type = 0;
    stream->mode = 0;
    stream->state = 0;
    stream->inb = stream->inr = stream->outb = stream->outr = 0;
    stream->tick_i = stream->tick_o = stream->tact = stream->inbt = stream->outbt = 0;
    //def_initlock(&stream->lock);
    stream->port = NULL;
    stream->path[0] = '\0';
    stream->msg[0] = '\0';
}
